//
//  RxSwiftOperatorsViewController.swift
//  DemoProject
//
//  Created by Hao on 2018/8/5.
//  Copyright © 2018年 hao. All rights reserved.
//

import UIKit
//import RxSwift
//import RxCocoa

/// 操作符:
/// 变换操作符：buffer、map、flatMap、scan等
/// 过滤操作符：filter、take、skip等
/// 条件和布尔操作符：amb、takeWhile、skipWhile等
/// 结合操作符：startWith、merge、zip等
/// 算数&聚合操作符：toArray、reduce、concat
/// 连接操作符：connect、publish、replay、multicast
/// 其他操作符：delay、materialize、timeout等
class RxSwiftOperatorsViewController: BaseViewController {

    let disposeBag = DisposeBag()
    
    override func viewDidLoad() {
        super.viewDidLoad()
//        testTransformingObservables()
//        testFilteringObservables()
//        testConditionalAndBooleanOperators()
//        testCombiningObservables()
//        testMathematicalAndAggregateOperators()
//        testConnectableObservableOperators()
        testObservableUtilityOperators()
    }
    
    fileprivate func sendNext(observable: PublishSubject<String>) {
        observable.onNext("a")
        observable.onNext("b")
        observable.onNext("c")

        observable.onNext("1")
        observable.onNext("2")
        observable.onNext("3")
    }

}

// MARK: 变换操作符：buffer、map、flatMap、scan等
extension RxSwiftOperatorsViewController {
    
    fileprivate func testTransformingObservables() {
//        _bufferOperators()
//        _windowOperators()
//        _mapOperators()
//        _flatMapOperators()
//        _flatMapLatestOperators()
//        _flatMapFirstOperators()
//        _concatMapOperators()
//        _scanOperators()
        _groupByOperators()
    }
    
    /// buffer 方法作用是缓冲组合，第一个参数是缓冲时间，第二个参数是缓冲个数，第三个参数是线程。
    /// 该方法简单来说就是缓存 Observable 中发出的新元素，当元素达到某个数量，或者经过了特定的时间，它就会将这个元素集合发送出来。
    /// 运行结果:
    ///     next(["a", "b", "c"])
    ///     next(["1", "2", "3"])
    ///     next([])
    ///     next([])
    private func _bufferOperators() {
        let subject = PublishSubject<String>()
        // 每缓存3个元素则组合起来一起发出。
        // 如果1秒钟内不够3个也会发出（有几个发几个，一个都没有发空数组 []
        subject
            .buffer(timeSpan: 2, count: 3, scheduler: MainScheduler.instance)
            .subscribe { print(info: $0) }
            .disposed(by: disposeBag)
        
        sendNext(observable: subject)
    }
    
    /// window 操作符和 buffer 十分相似。不过 buffer 是周期性的将缓存的元素集合发送出来，而 window 周期性的将元素集合以 Observable 的形态发送出来。
    /// 同时 buffer 要等到元素搜集完毕后，才会发出元素序列。而 window 可以实时发出元素序列
    /// 运行结果:
    ///     "subscribe: RxSwift.AddRef<Swift.String>"
    ///     "$0.subscribe: next(a)"
    ///     "$0.subscribe: next(b)"
    ///     "$0.subscribe: next(c)"
    ///     "$0.subscribe: completed"
    ///     "subscribe: RxSwift.AddRef<Swift.String>"
    ///     "$0.subscribe: next(1)"
    ///     "$0.subscribe: next(2)"
    ///     "$0.subscribe: next(3)"
    ///     "$0.subscribe: completed"
    ///     "subscribe: RxSwift.AddRef<Swift.String>"
    ///     "$0.subscribe: completed"
    private func _windowOperators() {
        let subject = PublishSubject<String>()
        
        //每3个元素作为一个子Observable发出。
        subject
            .window(timeSpan: 2, count: 3, scheduler: MainScheduler.instance)
            .subscribe(onNext: { [weak self] in
                print(info: "subscribe: \($0)")
                $0.asObservable().subscribe { print(info: "$0.subscribe: \($0)") }.disposed(by: self!.disposeBag)
            })
            .disposed(by: disposeBag)
        
        sendNext(observable: subject)
    }
    
    /// 该操作符通过传入一个函数闭包把原来的 Observable 序列转变为一个新的 Observable 序列。
    /// 运行结果:
    ///     map value: next(11)
    ///     map value: next(12)
    ///     map value: next(14)
    ///     map value: completed
    private func _mapOperators() {
        Observable.of(1, 2, 4).map { $0+10 }.subscribe { print(info: "map value: \($0)") }.disposed(by: disposeBag)
    }
    
    /// 当 Observable 的元素本生拥有其他的 Observable 时，可以将所有子 Observables 的元素发送出来
    private func _flatMapOperators() {
        let subject1 = BehaviorSubject<String>(value: "A")
        let subject2 = BehaviorSubject<String>(value: "1")
        
        let behaviorRelay = BehaviorRelay(value: subject1)
        /** 将 BehaviorSubject 对象携带的值做操作, 如果只有一行代码可以省略 ‘-> (BehaviorSubject<String>)’ */
        behaviorRelay.flatMap { subj -> (BehaviorSubject<String>) in
            if let value = try? subj.value() {
                subj.onNext("flatmap: " + value)
            }
            return subj
        } .subscribe{ print(info: "subscribe: \($0)") }.disposed(by: disposeBag)
        // 运行结果:
        // subscribe: next(flatmap: A)
        /** 简写 */
//        behaviorRelay.flatMap { $0 } .subscribe { print(info: "subscribe: \($0)") }.disposed(by: disposeBag)
        print(info: "-----------------")
        // 改变subject1的value并不会走flatmap的转换
        // 运行结果:
        // subscribe: next(B)
        subject1.onNext("B")
        print(info: "-----------------")
        
        // behaviorRelay.accept()方法后会走flatmap转换
        // 运行结果:
        // subscribe: next(flatmap: 1)
        // subscribe: next(2)
        behaviorRelay.accept(subject2)
        subject2.onNext("2")
        
        print(info: "-----------------")
        // 虽然赋值的是同一对象，但flatmap操作符会对源 Observable 的每一个元素应用一个转换方法，将他们转换成 Observables。 然后将这些 Observables 的元素合并之后再发送出来。即又将其 "拍扁"（降维）成一个 Observable 序列
        // 运行结果:
        // subscribe: next(flatmap: B)
        // subscribe: next(flatmap: B)
        // subscribe: next(C)
        // subscribe: next(C)
        behaviorRelay.accept(subject1)
        subject1.onNext("C")
        
//        let subject3 = BehaviorSubject<String>(value: "_a")
//        behaviorRelay.accept(subject3)
//        subject3.onNext("_b")
    }
    
    /// flatMapLatest 与 flatMap 的唯一区别是：flatMapLatest 只会接收最新的 value 事件
    /// 运行结果:
    ///     A
    ///     B
    ///     1
    ///     2
    private func _flatMapLatestOperators() {
        let subject1 = BehaviorSubject(value: "A")
        let subject2 = BehaviorSubject(value: "1")
        
        let relay = BehaviorRelay(value: subject1)
        
        relay
            .flatMapLatest { $0 }
            .subscribe(onNext: { print(info: $0) })
            .disposed(by: disposeBag)
        
        subject1.onNext("B")
        relay.accept(subject2)
        subject2.onNext("2")
        subject1.onNext("C")
    }
    
    /// flatMapFirst 与 flatMapLatest 正好相反：flatMapFirst 只会接收最初的 value 事件。
    /// 运行结果:
    ///     A
    ///     B
    ///     C
    private func _flatMapFirstOperators() {
        let subject1 = BehaviorSubject(value: "A")
        let subject2 = BehaviorSubject(value: "1")
        
        let relay = BehaviorRelay(value: subject1)
        
        relay
            .flatMapFirst { $0 }
            .subscribe(onNext: { print(info: $0) })
            .disposed(by: disposeBag)
        
        subject1.onNext("B")
        relay.accept(subject2)
        subject2.onNext("2")
        subject1.onNext("C")
    }
    
    /// concatMap 与 flatMap 的唯一区别是：当前一个 Observable 元素发送完毕后，后一个Observable 才可以开始发出元素。或者说等待前一个 Observable 产生完成事件后，才对后一个 Observable 进行订阅
    /// 运行结果:
    ///     A
    ///     B
    ///     C
    ///     2
    private func _concatMapOperators() {
        let subject1 = BehaviorSubject(value: "A")
        let subject2 = BehaviorSubject(value: "1")
        
        let relay = BehaviorRelay(value: subject1)
        
        relay
            .concatMap { $0 }
            .subscribe(onNext: { print(info: $0) })
            .disposed(by: disposeBag)
        
        subject1.onNext("B")
        relay.accept(subject2)
        subject2.onNext("2")
        subject1.onNext("C")
        
        /// 只有前一个序列结束，才能接收下一个序列
        subject1.onCompleted()
    }
    
    /// scan 就是先给一个初始化的数，然后不断的拿前一个结果和最新的值进行处理操作
    /// 运行结果:
    ///    acum -> 10, elem -> 1
    ///    next(11)
    ///    acum -> 11, elem -> 2
    ///    next(13)
    ///    acum -> 13, elem -> 3
    ///    next(16)
    ///    acum -> 16, elem -> 4
    ///    next(20)
    ///    acum -> 20, elem -> 5
    ///    next(25)
    ///    completed
    private func _scanOperators() {
        // scan()方法的参数就是初始值
        Observable.of(1, 2, 3, 4, 5).scan(10) { (acum, elem) -> Int in
            // acum: 上一个结果(第一个就是初始值)  elem: 序列中的值
            print(info: "acum -> \(acum), elem -> \(elem)")
            return acum + elem
        }.subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// groupBy 操作符将源 Observable 分解为多个子 Observable，然后将这些子 Observable 发送出来。
    /// 该操作符会将元素通过 '某个键' 进行分组，然后将分组后的元素序列以 Observable 的形态发送出来
    /// 运行结果:
    ///    key：偶数 event：0
    ///    key：奇数 event：1
    ///    key：偶数 event：2
    ///    key：奇数 event：3
    ///    key：偶数 event：4
    ///    key：奇数 event：5
    ///    ----> end
    private func _groupByOperators() {
        Observable<Int>.of(0, 1, 2, 3, 4, 5).groupBy { (element) -> String in
            return element%2 == 0 ? "偶数" : "奇数"
        }.subscribe { (event) in
            switch event {
                case .next(let group):
                    group.asObservable().subscribe(onNext: { (event) in
                        print(info: "key：\(group.key) event：\(event)")
                    }).disposed(by: self.disposeBag)
                default:
                    print(info: "----> end")
            }
        }.disposed(by: disposeBag)
    }
    
}

// MARK: 过滤操作符：filter、take、skip等
extension RxSwiftOperatorsViewController {
    
    fileprivate func testFilteringObservables() {
//        _filterOperators()
//        _distinctUntilChangedOperators()
//        _singleOperators()
//        _elementAtOperators()
//        _ignoreElementsOperators()
//        _takeOperators()
//        _takeLastOperators()
//        _skipOperators()
//        _sampleOperators()
        _debounceOperators()
    }
    
    /// 该操作符就是用来过滤掉某些不符合要求的事件
    /// 运行结果:
    ///    next(30)
    ///    next(22)
    ///    next(60)
    ///    next(40)
    ///    completed
    private func _filterOperators() {
        Observable.of(2, 30, 22, 5, 60, 3, 40, 9).filter { (value) -> Bool in
            value > 10  // 返回所有大于10的值
        }.subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该操作符用于过滤掉连续重复的事件
    /// 运行结果:
    ///    next(1)
    ///    next(2)
    ///    next(4)
    ///    next(5)
    ///    next(2)
    ///    next(3)
    ///    completed
    private func _distinctUntilChangedOperators() {
        Observable.of(1, 2, 4, 4, 5, 2, 3, 3).distinctUntilChanged().subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 限制只发送一次事件，或者满足条件的第一个事件
    /// 如果存在有多个事件或者没有事件都会发出一个 error 事件
    /// 如果只有一个事件，则不会发出 error 事件
    /// 运行结果:
    ///    next(2)
    ///    completed
    ///    next(A)
    ///    error(Sequence contains more than one element.)
    private func _singleOperators() {
        Observable.of(1, 2, 3, 4).single { (value) -> Bool in
            value == 2
        }.subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        Observable.of("A", "B", "C", "D").single().subscribe { print(info: $0) }.disposed(by: disposeBag)
//        Observable.of("111").single().subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该方法实现只处理在指定位置的事件
    /// 运行结果:
    ///    next(3)
    ///    completed
    private func _elementAtOperators() {
        // 只处理第二个位置的事件(从0开始)
        Observable.of(1, 2, 3, 4).elementAt(2).subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该操作符可以忽略掉所有的元素，只发出 error 或 completed 事件。
    /// 如果我们并不关心 Observable 的任何元素，只想知道 Observable 在什么时候终止，那就可以使用 ignoreElements 操作符
    /// 运行结果:
    ///    completed
    private func _ignoreElementsOperators() {
        Observable.of(1, 2, 3, 4).ignoreElements().subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该方法实现仅发送 Observable 序列中的前 n 个事件，在满足数量之后会自动 .completed
    /// 运行结果:
    ///    1
    ///    2
    private func _takeOperators() {
        // 只发送前2个事件就自动发送.completed
        Observable.from([1, 2, 3, 4]).take(2).subscribe(onNext: { print(info: $0) }).disposed(by: disposeBag)
    }
    
    /// 该方法实现仅发送 Observable 序列中的后 n 个事件
    /// 运行结果:
    ///    2
    ///    3
    ///    4
    private func _takeLastOperators() {
        Observable.of(1, 2, 3, 4).takeLast(3).subscribe(onNext: { print(info: $0) }).disposed(by: disposeBag)
    }
    
    /// 该方法用于跳过源 Observable 序列发出的前 n 个事件
    /// 运行结果:
    ///    3
    ///    4
    private func _skipOperators() {
        Observable.of(1, 2, 3, 4).skip(2).subscribe(onNext: { print(info: $0) }).disposed(by: disposeBag)
    }
    
    /// Sample 除了订阅源 Observable 外，还可以监视另外一个 Observable， 即 notifier
    /// 每当收到 notifier 事件，就会从源序列取一个最新的事件并发送。而如果两次 notifier 事件之间没有源序列的事件，则不发送值
    /// 运行结果:
    ///    1
    ///    2
    ///    4
    ///    5
    private func _sampleOperators() {
        let source = PublishSubject<Int>()
        let notifier = PublishSubject<String>()
        
        source.sample(notifier).subscribe(onNext: { print(info: $0) }).disposed(by: disposeBag)
        source.onNext(1)
        
        // 让源序列接收消息
        notifier.onNext("A")
        
        source.onNext(2)
        
        // 让源序列接收消息
        notifier.onNext("B")
        notifier.onNext("C")
        
        source.onNext(3)
        source.onNext(4)
        
        // 让源序列接收消息
        notifier.onNext("D")
        
        source.onNext(5)
        
        notifier.onCompleted()
    }
    
    /// debounce 操作符可以用来过滤掉高频产生的元素，它只会发出这种元素：该元素产生后，一段时间内没有新元素产生。
    /// 即队列中的元素如果和下一个元素的间隔小于了指定的时间间隔，那么这个元素将被过滤掉。
    /// debounce 常用在用户输入的时候，不需要每个字母敲进去都发送一个事件，而是稍等一下取最后一个事件
    /// 运行结果:
    ///    2018-08-06 23:08:40.222 flatMap -> item: ["time": 0.10000000000000001, "value": 1.0]
    ///    2018-08-06 23:08:40.222 flatMap -> item: ["time": 1.1000000000000001, "value": 2.0]
    ///    2018-08-06 23:08:40.223 flatMap -> item: ["time": 1.2, "value": 3.0]
    ///    2018-08-06 23:08:40.223 flatMap -> item: ["time": 1.2, "value": 4.0]
    ///    2018-08-06 23:08:40.223 flatMap -> item: ["time": 1.3999999999999999, "value": 5.0]
    ///    2018-08-06 23:08:40.224 flatMap -> item: ["time": 2.1000000000000001, "value": 6.0]
    ///    2018-08-06 23:08:41.324 next(1)
    ///    2018-08-06 23:08:42.325 next(6)
    ///    2018-08-06 23:08:42.326 completed
    private func _debounceOperators() {
        // 定义好每个事件里的值以及发送的时间
        let times = [
            [ "value": 1, "time": 0.1 ],
            [ "value": 2, "time": 1.1 ],
            [ "value": 3, "time": 1.2 ],
            [ "value": 4, "time": 1.2 ],
            [ "value": 5, "time": 1.4 ],
            [ "value": 6, "time": 2.1 ]
        ]
        // 生成对应的 Observable 序列并订阅
        Observable.from(times).flatMap { item -> Observable<Int> in
            // delaySubscription: 延迟订阅
            print(info: "flatMap -> item: \(item)")
            return Observable.of(Int(item["value"]!)).delaySubscription(item["time"]!, scheduler: MainScheduler.instance)
        }.debounce(1, scheduler: MainScheduler.instance) // //只发出与下一个间隔超过 0.5秒 的元素
         .subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
}

// MARK: 条件和布尔操作符：amb、takeWhile、skipWhile等
extension RxSwiftOperatorsViewController {
    
    fileprivate func testConditionalAndBooleanOperators() {
//        _ambOperators()
//        _takeWhileOperators()
//        _takeUntilOperators()
//        _skipWhileOperators()
        _skipUntilOperators()
    }
    
    /// 当传入多个 Observables 到 amb 操作符时，它将取第一个发出元素或产生事件的 Observable，然后只发出它的元素。并忽略掉其他的 Observables
    /// 运行结果:
    ///    next(1)
    ///    next(2)
    ///    next(3)
    private func _ambOperators() {
        let subject1 = PublishSubject<Int>()
        let subject2 = PublishSubject<Int>()
        let subject3 = PublishSubject<Int>()
        /// 多次调用amb()方法的参数并不影响， subject发送.next事件先后才影响， 最先发送的才会发出，其他的忽略
        subject1.amb(subject3).amb(subject2).subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject2.onNext(1)
        subject3.onNext(0)
        subject1.onNext(20)
        subject2.onNext(2)
        subject1.onNext(40)
        subject2.onNext(3)
        subject1.onNext(60)
        subject3.onNext(0)
        subject3.onNext(0)
    }
    
    /// 该方法依次判断 Observable 序列的每一个值是否满足给定的条件。 当第一个不满足条件的值出现时，它便自动完成
    /// 运行结果:
    ///     completed
    private func _takeWhileOperators() {
        Observable.of(1, 2, 3, 4).takeWhile({ (value) -> Bool in
            print(info: "value: \(value)")
            return value%2 == 0
        }) .subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 除了订阅源 Observable 外，通过 takeUntil 方法我们还可以监视另外一个 Observable， 即 notifier。
    /// 如果 notifier 发出值或 complete 通知，那么源 Observable 便自动完成，停止发送事件
    /// 运行结果:
    ///    next(A)
    ///    next(B)
    ///    completed
    private func _takeUntilOperators() {
        let source = PublishSubject<String>()
        let notifier = PublishSubject<String>()
        source.takeUntil(notifier).subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        source.onNext("A")
        source.onNext("B")
        
        // 停止接收消息
        notifier.onNext("STOP")
        
        source.onNext("C")
        source.onNext("D")
    }
    
    /// 该方法用于跳过前面所有满足条件的事件。 一旦遇到不满足条件的事件，之后就不会再跳过
    /// 运行结果:
    ///    next(3)
    ///    next(4)
    ///    completed
    private func _skipWhileOperators() {
        Observable.of(1, 2, 3, 4).skipWhile { $0 < 3 }.subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 同上面的 takeUntil 一样，skipUntil 除了订阅源 Observable 外，通过 skipUntil 方法我们还可以监视另外一个 Observable， 即 notifier 。
    /// 与 takeUntil 相反的是。源 Observable 序列事件默认会一直跳过，直到 notifier 发出值或 complete 通知
    /// 运行结果:
    ///    next(C)
    ///    next(D)
    ///    next(E)
    private func _skipUntilOperators() {
        let source = PublishSubject<String>()
        let notifier = PublishSubject<String>()
        source.skipUntil(notifier).subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        source.onNext("A")
        source.onNext("B")
        
        // 开始接收消息
        notifier.onNext("START")
        
        source.onNext("C")
        source.onNext("D")
        
        // 继续接收消息
        notifier.onNext("NEXT")
        
        source.onNext("E")
    }
}

// MARK: 结合操作符：startWith、merge、zip等
extension RxSwiftOperatorsViewController {
    
    fileprivate func testCombiningObservables() {
//        _startWithOperators()
//        _mergeOperators()
//        _zipOperators()
//        _combineLatestOperators()
//        _withLatestFromOperators()
        _switchLatestOperators()
    }
    
    /// 该方法会在 Observable 序列开始之前插入一些事件元素。即发出事件消息之前，会先发出这些预先插入的事件消息 (类似插队)
    /// 运行结果:
    ///    next(0)
    ///    next(1)
    ///    next(2)
    ///    next(3)
    ///    next(4)
    ///    completed
    private func _startWithOperators() {
        Observable.of(3, 4).startWith(1, 2).startWith(0).subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该方法可以将多个（两个或两个以上的）Observable 序列合并成一个 Observable 序列
    /// 运行结果:
    ///    next(20)
    ///    next(40)
    ///    next(1)
    ///    next(100)
    ///    next(2)
    ///    completed
    private func _mergeOperators() {
        let subject1 = PublishSubject<Int>()
        let subject2 = PublishSubject<Int>()
        
        Observable.merge(subject1, subject2).subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject1.onNext(20)
        subject1.onNext(40)
        subject2.onNext(1)
        subject1.onNext(100)
        subject2.onNext(2)
        
        // 必须都结束才会打印 completed
        subject1.onCompleted()
        subject2.onCompleted()
    }
    
    /// 该方法可以将多个（两个或两个以上的）Observable 序列压缩成一个 Observable 序列 (zip 常常用在整合网络请求上)
    /// 而且它会等到每个 Observable 事件一一对应地凑齐之后再合并
    /// 运行结果:
    ///    next(zip -> 1A)
    ///    next(zip -> 2B)
    ///    next(zip -> 3C)
    ///    completed
    private func _zipOperators() {
        let subject1 = PublishSubject<Int>()
//        let subject2 = PublishRelay<String>()
        let subject2 = PublishSubject<String>()
        Observable.zip(subject1, subject2) { (s1, s2) -> String in
//            print(info: "Thread -> \(Thread.current)")
            return "zip -> \(s1)" + s2
        }.subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject1.onNext(1)
        subject2.onNext("A")
        subject1.onNext(2)
        subject2.onNext("B")
        subject2.onNext("C")
        subject1.onNext(3)
        
        subject1.onNext(4)
        // 必须都结束才会打印 completed
        subject1.onCompleted()
        subject2.onCompleted()
    }
    
    /// 该方法同样是将多个（两个或两个以上的）Observable 序列元素进行合并。
    /// 但与 zip 不同: 每当任意一个 Observable 有新的事件发出时，它会将每个 Observable 序列的最新的一个事件元素进行合并
    /// 运行结果:
    ///    next(1A)
    ///    next(2A)
    ///    next(2B)
    ///    next(2C)
    ///    next(3C)
    ///    next(4C)
    private func _combineLatestOperators() {
        let subject1 = PublishSubject<Int>()
        let subject2 = PublishSubject<String>()
        Observable.combineLatest(subject1, subject2) { "\($0)\($1)" }
            .subscribe { print(info: $0) }
            .disposed(by: disposeBag)
        
        subject1.onNext(1)
        subject2.onNext("A")
        subject1.onNext(2)
        subject2.onNext("B")
        subject2.onNext("C")
        subject1.onNext(3)
        subject1.onNext(4)
    }
    
    /// 将两个 Observable 序列合并为一个。每当 self 队列发射一个元素时，便从第二个序列中取出最新的一个值
    /// 运行结果:
    ///    next(A)
    ///    next(C)
    private func _withLatestFromOperators() {
        let subject1 = PublishSubject<String>()
        let subject2 = PublishSubject<String>()
        subject2.withLatestFrom(subject1).subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject1.onNext("A")
        subject2.onNext("1")
        subject1.onNext("B")
        subject1.onNext("C")
        subject2.onNext("2")
        subject1.onNext("D")
    }
    
    /// switchLatest 有点像其他语言的 switch 方法，可以对事件流进行转换。
    /// 比如本来监听的 subject1，通过更改 variable 里面的 value 更换事件源。变成监听 subject2
    /// 运行结果:
    ///    next(B)
    ///    next(C)
    ///    next(2)
    ///    next(E)
    private func _switchLatestOperators() {
        let subject1 = PublishSubject<String>()
        let subject2 = PublishSubject<String>()
        // 监听 subject1 事件源
        let behaviorRelay = BehaviorRelay.init(value: subject1)
        behaviorRelay.switchLatest().subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject1.onNext("B")
        subject1.onNext("C")
        
        //改变事件源
        behaviorRelay.accept(subject2)
        subject1.onNext("D")
        subject2.onNext("2")
        
        //改变事件源
        behaviorRelay.accept(subject1)
        subject2.onNext("3")
        subject1.onNext("E")
    }
    
}

// MARK: 算数&聚合操作符：toArray、reduce、concat
extension RxSwiftOperatorsViewController {
    
    fileprivate func testMathematicalAndAggregateOperators() {
//        _toArrayOperators()
//        _reduceOperators()
        _concatOperators()
    }
    
    /// 该操作符先把一个序列转成一个数组，并作为一个单一的事件发送，然后结束
    /// 运行结果:
    ///    next([1, 2, 3, 4])
    ///    completed
    private func _toArrayOperators() {
        Observable.of(1, 2, 3, 4).toArray().subscribe { print(info: $0) }.dispose()
    }
    
    /// reduce 接受一个初始值，和一个操作符号。
    /// reduce 将给定的初始值，与序列里的每个值进行累计运算。得到一个最终结果，并将其作为单个值发送出去
    /// 运行结果:
    ///    next(28)
    ///    completed
    private func _reduceOperators() {
        // 方法 reduce() 的参数可简写为: reduce(10, accumulator: +)
        Observable.of(1, 2, 6, 9).reduce(10, accumulator: { $0 + $1 }).subscribe { print(info: $0) }.dispose()
    }
    
    /// concat 会把多个 Observable 序列合并（串联）为一个 Observable 序列。
    /// 并且只有当前面一个 Observable 序列发出了 completed 事件，才会开始发送下一个 Observable 序列事件
    /// 运行结果:
    ///    next(1)
    ///    next(1)
    ///    next(1)
    ///    next(3)
    ///    next(3)
    private func _concatOperators() {
        let subject1 = BehaviorSubject(value: 1)
        let subject2 = BehaviorSubject(value: 2)
        
        let variable = BehaviorRelay(value: subject1)
        variable.asObservable().concat().subscribe { print(info: $0) }.disposed(by: disposeBag)
        
        subject2.onNext(3)
        subject1.onNext(1)
        subject1.onNext(1)
        subject1.onCompleted()
        // subject1结束事件，还得切换监听subject2的事件,不然收不到事件
        variable.accept(subject2)
        subject2.onNext(3)
    }
    
}

// MARK: 连接操作符：connect、publish、replay、multicast
extension RxSwiftOperatorsViewController {

    fileprivate func testConnectableObservableOperators() {
//        _testoOrdinaryObservable()
//        _publishOperators()
//        _replayOperators()
//        _multicastOperators()
//        _refCountOperators()
        _shareOperators()
    }
    
    ///延迟执行
    /// - Parameters:
    ///   - delay: 延迟时间（秒）
    ///   - closure: 延迟执行的闭包
    private func _delayRun(_ delay: Double, closure: @escaping () -> Void) {
        DispatchQueue.main.asyncAfter(deadline: .now() + delay) {
            closure()
        }
    }
    
    ///    运行结果:
    ///    2018-08-08 09:38:11.467 订阅者1: next(0)
    ///    2018-08-08 09:38:12.467 订阅者1: next(1)
    ///    2018-08-08 09:38:13.467 订阅者1: next(2)
    ///    2018-08-08 09:38:14.467 订阅者1: next(3)
    ///    2018-08-08 09:38:15.467 订阅者1: next(4)
    ///    2018-08-08 09:38:16.466 订阅者1: next(5)
    ///    2018-08-08 09:38:16.467 订阅者2: next(0)
    ///    2018-08-08 09:38:17.467 订阅者1: next(6)
    ///    2018-08-08 09:38:17.468 订阅者2: next(1)
    ///    2018-08-08 09:38:18.466 订阅者1: next(7)
    ///    2018-08-08 09:38:18.467 订阅者2: next(2)
    private func _testoOrdinaryObservable() {
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        // 第一个订阅者(立即开始订阅)
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        //第二个订阅者（延迟5秒开始订阅)
        _delayRun(5) {
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }

    /// 可连接序列
    /// publish 方法会将一个正常的序列转换成一个可连接的序列。同时该序列不会立刻发送事件，只有在调用 connect 之后才会开始
    /// 运行结果:
    ///    2018-08-08 09:51:28.516  [805]:  START ------
    ///    2018-08-08 09:51:28.517  [811]:  -----------
    ///    2018-08-08 09:51:32.519  [809]:  订阅者1: next(0)
    ///    2018-08-08 09:51:33.519  [809]:  订阅者1: next(1)
    ///    2018-08-08 09:51:34.519  [809]:  订阅者1: next(2)
    ///    2018-08-08 09:51:35.519  [809]:  订阅者1: next(3)
    ///    2018-08-08 09:51:36.518  [809]:  订阅者1: next(4)
    ///    2018-08-08 09:51:36.519  [819]:  订阅者2: next(4)
    ///    2018-08-08 09:51:37.518  [809]:  订阅者1: next(5)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者2: next(5)
    ///    2018-08-08 09:51:38.519  [809]:  订阅者1: next(6)
    ///    2018-08-08 09:51:38.519  [819]:  订阅者2: next(6)
    private func _publishOperators() {
        print(info: "START ------ ")
        //每隔1秒钟发送1个事件
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance).publish()
        //第一个订阅者（立刻开始订阅）
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        
        print(info: "-----------")
        //相当于把事件消息推迟了3秒
        _delayRun(3) {
            // 调用connect()后才发送事件
            _ = interval.connect()
        }
    
        //第二个订阅者（延迟8秒开始订阅）
        _delayRun(8) {
            // 会与 第一个订阅者 的值同步 (publish操作符)
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }

    /// replay 同上面的 publish 方法相同之处在于：会将将一个正常的序列转换成一个可连接的序列。同时该序列不会立刻发送事件，只有在调用 connect 之后才会开始。
    /// replay 与 publish 不同在于：新的订阅者还能接收到订阅之前的事件消息（数量由设置的 bufferSize 决定）
    /// 运行结果:
    ///    2018-08-08 09:51:28.516  [805]:  START ------
    ///    2018-08-08 09:51:28.517  [811]:  -----------
    ///    2018-08-08 09:51:32.519  [809]:  订阅者1: next(0)
    ///    2018-08-08 09:51:33.519  [809]:  订阅者1: next(1)
    ///    2018-08-08 09:51:34.519  [809]:  订阅者1: next(2)
    ///    2018-08-08 09:51:35.519  [809]:  订阅者1: next(3)
    ///    2018-08-08 09:51:36.518  [809]:  订阅者1: next(4)
    ///    2018-08-08 09:51:36.519  [819]:  订阅者1: next(5)
    ///    2018-08-08 09:51:37.518  [809]:  订阅者1: next(6)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者2: next(4)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者2: next(5)
    ///    2018-08-08 09:51:38.519  [809]:  订阅者2: next(6)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者1: next(7)
    ///    2018-08-08 09:51:38.519  [819]:  订阅者2: next(7)
    private func _replayOperators() {
        print(info: "START ------ ")
        // replay(3): 能接收到订阅之前的3个事件消息
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance).replay(3)
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        print(info: "-----------")
        //第一个订阅者（延迟2秒开始订阅）
        _delayRun(2) {
            _ = interval.connect()
        }
        //第二个订阅者（延迟10秒开始订阅）
        _delayRun(10) {
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }

    /// multicast 方法同样是将一个正常的序列转换成一个可连接的序列。
    /// 同时 multicast 方法还可以传入一个 Subject，每当序列发送事件时都会触发这个 Subject 的发送
    /// 运行结果:
    ///    2018-08-08 10:44:43.836  [874]:  START ------
    ///    2018-08-08 10:44:43.837  [880]:  -----------
    ///    2018-08-08 10:44:46.838  [876]:  Subject: next(0)
    ///    2018-08-08 10:44:46.840  [879]:  订阅者1: next(0)
    ///    2018-08-08 10:44:47.838  [876]:  Subject: next(1)
    ///    2018-08-08 10:44:47.839  [879]:  订阅者1: next(1)
    ///    2018-08-08 10:44:48.838  [876]:  Subject: next(2)
    ///    2018-08-08 10:44:48.839  [879]:  订阅者1: next(2)
    ///    2018-08-08 10:44:49.838  [876]:  Subject: next(3)
    ///    2018-08-08 10:44:49.838  [879]:  订阅者1: next(3)
    ///    2018-08-08 10:44:49.839  [886]:  订阅者2: next(3)
    private func _multicastOperators() {
        print(info: "START ------ ")
        let subject = PublishSubject<Int>()
        _ = subject.subscribe { print(info: "Subject: \($0)") }
        
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance).multicast(subject)
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        print(info: "-----------")
        //第一个订阅者（延迟2秒开始订阅）
        _delayRun(2) {
            _ = interval.connect()
        }
        //第二个订阅者（延迟6秒开始订阅）
        _delayRun(6) {
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }
    
    /// refCount 操作符可以将可被连接的 Observable 转换为普通 Observable
    /// 即该操作符可以自动连接和断开可连接的 Observable。当第一个观察者对可连接的 Observable 订阅时，那么底层的 Observable 将被自动连接。当最后一个观察者离开时，那么底层的 Observable 将被自动断开连接
    /// 运行结果:
    ///    2018-08-08 09:51:28.516  [805]:  START ------
    ///    2018-08-08 09:51:32.519  [809]:  订阅者1: next(0)
    ///    2018-08-08 09:51:33.519  [809]:  订阅者1: next(1)
    ///    2018-08-08 09:51:34.519  [809]:  订阅者1: next(2)
    ///    2018-08-08 09:51:35.519  [809]:  订阅者1: next(3)
    ///    2018-08-08 09:51:36.518  [809]:  订阅者1: next(4)
    ///    2018-08-08 09:51:36.519  [819]:  订阅者1: next(5)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者2: next(5)
    ///    2018-08-08 09:51:37.519  [819]:  订阅者1: next(6)
    ///    2018-08-08 09:51:38.519  [809]:  订阅者2: next(6)
    private func _refCountOperators() {
        print(info: "START ------ ")
        // 将publish()后可被连接的 Observable 转换为普通 Observable，不需要.connect()即可订阅
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance).publish().refCount()
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        //第二个订阅者（延迟5秒开始订阅）
        _delayRun(5) {
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }
    
    /// 该操作符将使得观察者共享源 Observable，并且缓存最新的 n 个元素，将这些元素直接发送给新的观察者。
    /// 简单来说 shareReplay 就是 replay 和 refCount 的组合
    /// 运行结果:
    ///    2018-08-08 11:05:06.501  [933]:  START ------
    ///    2018-08-08 11:05:07.503  [938]:  订阅者1: next(0)
    ///    2018-08-08 11:05:08.503  [938]:  订阅者1: next(1)
    ///    2018-08-08 11:05:09.502  [938]:  订阅者1: next(2)
    ///    2018-08-08 11:05:10.503  [938]:  订阅者1: next(3)
    ///    2018-08-08 11:05:11.503  [938]:  订阅者1: next(4)
    ///    2018-08-08 11:05:11.504  [942]:  订阅者2: next(2)
    ///    2018-08-08 11:05:11.504  [942]:  订阅者2: next(3)
    ///    2018-08-08 11:05:11.504  [942]:  订阅者2: next(4)
    ///    2018-08-08 11:05:12.502  [938]:  订阅者1: next(5)
    ///    2018-08-08 11:05:12.503  [942]:  订阅者2: next(5)
    private func _shareOperators() {
        print(info: "START ------ ")
        //每隔1秒钟发送1个事件
        let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance).share(replay: 3)
        
        //第一个订阅者（立刻开始订阅）
        _ = interval.subscribe { print(info: "订阅者1: \($0)") }
        
        //第二个订阅者（延迟5秒开始订阅）
        _delayRun(5) {
            _ = interval.subscribe { print(info: "订阅者2: \($0)") }
        }
    }

}

// MARK: 其他操作符：delay、materialize、timeout等
extension RxSwiftOperatorsViewController {
    
    fileprivate func testObservableUtilityOperators() {
//        _delayOperators()
//        _delaySubscriptionOperators()
//        _materializeOperators()
//        _dematerializeOperators()
//        _timeoutOperators()
        _usingOperators()
    }
    
    /// 该操作符会将 Observable 的所有元素都先拖延一段设定好的时间，然后才将它们发送出来
    /// 运行结果:
    ///    next(1)
    ///    next(2)
    ///    next(3)
    ///    completed
    private func _delayOperators() {
        /// 延时2秒发出
        Observable.of(1, 2, 3).delay(2, scheduler: MainScheduler.instance).subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 使用该操作符可以进行延时订阅。即经过所设定的时间后，才对 Observable 进行订阅操作
    /// 运行结果:
    ///    next(1)
    ///    next(2)
    ///    next(3)
    ///    completed
    private func _delaySubscriptionOperators() {
        Observable.of(1, 2, 3).delaySubscription(2, scheduler: MainScheduler.instance).subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该操作符可以将序列产生的事件，转换成元素。
    /// 通常一个有限的 Observable 将产生零个或者多个 onNext 事件，最后产生一个 onCompleted 或者 onError 事件。而 materialize 操作符会将 Observable 产生的这些事件全部转换成元素，然后发送出来
    /// 运行结果:
    ///    next(next(1))
    ///    next(next(2))
    ///    next(next(3))
    ///    next(completed)
    ///    completed
    private func _materializeOperators() {
        Observable.of(1, 2, 3).materialize().subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 该操作符的作用和 materialize 正好相反，它可以将 materialize 转换后的元素还原
    /// 运行结果:
    ///    next(1)
    ///    next(2)
    ///    next(3)
    ///    completed
    private func _dematerializeOperators() {
        Observable.of(1, 2, 3).materialize().dematerialize().subscribe { print(info: $0) }.disposed(by: disposeBag)
    }
    
    /// 使用该操作符可以设置一个超时时间。如果源 Observable 在规定时间内没有发任何出元素，就产生一个超时的 error 事件
    /// 运行结果:
    ///    1
    ///    2
    ///    3
    ///    Sequence timeout.
    private func _timeoutOperators() {
        //定义好每个事件里的值以及发送的时间
        let times = [
            [ "value": 1, "time": 0 ],
            [ "value": 2, "time": 0.5 ],
            [ "value": 3, "time": 1.5 ],
            [ "value": 4, "time": 4 ],
            [ "value": 5, "time": 5 ]
        ]
        
        //生成对应的 Observable 序列并订阅
        Observable.from(times)
            .flatMap { item in
                return Observable.of(Int(item["value"]!))
                    .delaySubscription(Double(item["time"]!),
                                       scheduler: MainScheduler.instance)
            }
            .timeout(2, scheduler: MainScheduler.instance) //超过两秒没发出元素，则产生error事件
            .subscribe(onNext: { element in
                print(info: element)
            }, onError: { error in
                print(info: error)
            })
            .disposed(by: disposeBag)
    }
    
    /// 使用 using 操作符创建 Observable 时，同时会创建一个可被清除的资源，一旦 Observable 终止了，那么这个资源就会被清除掉
    /// 运行结果:
    ///    开始订阅 infinite
    ///    开始订阅 limited
    ///    infinite: 0
    ///    infinite: 1
    ///    infinite: 2
    ///    infinite: 3
    ///    infinite: 4
    ///    limited: 0
    ///    infinite: 5
    ///    infinite: 6
    ///    infinite: 7
    ///    infinite: 8
    ///    infinite: 9
    ///    limited: 1
    ///    销毁 limited
    ///    销毁 infinite
    private func _usingOperators() {
        //一个无限序列（每隔0.1秒创建一个序列数
        let infiniteInterval = Observable<Int>.interval(0.1, scheduler: MainScheduler.instance)
            .do(onNext: { print(info: "infinite: \($0)") },
                onSubscribe: { print(info: "开始订阅 infinite")},
                onDispose: { print(info: "销毁 infinite")})
        //一个有限序列（每隔2秒创建一个序列数，共创建三个 ）
        let limited = Observable<Int>
            .interval(0.5, scheduler: MainScheduler.instance)
            .take(2)
            .do(
                onNext: { print(info: "limited: \($0)") },
                onSubscribe: { print(info: "开始订阅 limited")},
                onDispose: { print(info: "销毁 limited")}
            )
        
        // 使用using操作符创建序列
        let o = Observable<Int>.using({ () -> TestUsingDisposable in
            return TestUsingDisposable(infiniteInterval.subscribe())
        }, observableFactory: { _ -> Observable<Int> in
            return limited
        })
        _ = o.subscribe()
    }
    
}

fileprivate class TestUsingDisposable: Disposable {
    let _dispose: () -> Void
    
    init(_ disposable: Disposable) {
        _dispose = disposable.dispose
    }
    
    func dispose() {
        _dispose()
    }
}

